package com.dyh.quickdemozwei.futureopreate;


import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

@Slf4j
public abstract class AbstractOperateInspector {
    protected BlockingQueue<FutureOperate> queue;
    protected BlockingQueue<GroupFutureOperate>  watchingGroups;
    protected WorkerThread[] workerThreads;
    protected MasterThread masterThread;
    protected int drainNum = 5;


    private final static boolean isPowerOfTwo(int val) {
        return (val & -val) == val;
    }

//    private final static void checkPowerOfTwo(int val) {
//
//    }


    public void setQueue(BlockingQueue<FutureOperate> queue) {
        this.queue = queue;
    }

    public void setWatchingGroups(BlockingQueue<GroupFutureOperate> watchingGroups) {
        this.watchingGroups = watchingGroups;
    }

    public void setWorkerThreads(WorkerThread[] workerThreads) {
        if (!isPowerOfTwo(workerThreads.length))
            throw new RuntimeException("must power of 2!");
        this.workerThreads = workerThreads;
    }

    class WorkerThread extends Thread{
//        final private  FutureOperate[] watchings;

        private final BlockingQueue<FutureOperate> watchings;
        private int watchingNum;
        int busy;

        public WorkerThread(int watchingNum) {
            this.watchingNum = watchingNum;
//            this.watchings = new FutureOperate[watchingNum];
            this.watchings = new LinkedBlockingQueue<>(watchingNum);
        }

        //不能阻塞
        boolean add(FutureOperate futureOperate){
            if (watchings.offer(futureOperate)){
                busy++;
                return true;
            }
            return false;

        }


        @Override
        public void run() {
            for (;;){
                //先peek 防止被塞入工作
                FutureOperate futureOperate = watchings.peek();
                if (futureOperate==null)
                    continue;
                if (futureOperate.finished())
                    watchings.poll();
                else {
                    futureOperate = watchings.poll();
                    if (futureOperate==null)
                        continue;
                    if (futureOperate.finished())
                        continue;
                    futureOperate.work(null);
                    futureOperate.onFinished();
                    busy--;
                }

            }
        }

    }

 /*
 *
for (int i = 0;i<watchings.length;i++){
                    if (watchings[i]==null)
                        continue;
                    if (watchings[i].finished()){
                        watchings[i].onFinished();
                        watchings[i]=null;
                        busy--;
                    }
                }
 */


    private void assignWorker(int index,FutureOperate operate){
        int  i = index & workerThreads.length-1;


        for (;;){
            WorkerThread workerThread = workerThreads[i];
            if (workerThread==null)
                continue;
            if (workerThread.busy < workerThread.watchingNum){
                workerThread.add(operate);
                break;
            }
            i = i+1 & workerThreads.length-1;
        }


    }

    class MasterThread extends Thread{
        @Override
        public void run() {
            int loop=0;
            for (;;loop++){
                try {
//                    FutureOperate futureOperate =  queue.take();

                    if (queue.size()>0){
                        List<FutureOperate> futureOperates = new ArrayList<>(drainNum);
                        queue.drainTo(futureOperates,drainNum);


                        for (int i = 0; i < futureOperates.size(); i++) {
                            FutureOperate operate = futureOperates.get(i);
                            if (operate instanceof GroupFutureOperate){
                                GroupFutureOperate groupFutureOperate = (GroupFutureOperate)operate;
                                for (int j = 0; j < groupFutureOperate.futureOperates.size(); j++) {
                                    assignWorker(j, groupFutureOperate.futureOperates.get(j));
                                }
                                watchingGroups.put(groupFutureOperate);
                            }
                            else
                                assignWorker(i,operate);
                        }

                    }




                    if (!watchingGroups.isEmpty()){
                        for (int i = 0; i < watchingGroups.size(); i++) {
                            GroupFutureOperate groupFutureOperate = watchingGroups.peek();
                            if (groupFutureOperate.finished()){
                                groupFutureOperate.onFinished();
                                watchingGroups.poll();
                            }
                        }

                    }





                }
                // catch (InterruptedException e) {
//                    break;
//                }

                catch (InterruptedException e) {

                } finally {

                }


            }
        }
    }

    public void start(){
        masterThread = new MasterThread();
        masterThread.setName("master");
        for (int i = 0; i < workerThreads.length; i++) {
            if (workerThreads[i]==null){
                WorkerThread workerThread = new WorkerThread(5);
                workerThread.setName("worker-"+i);
                workerThreads[i]= workerThread;
                workerThread.start();
            }

        }
        masterThread.start();
    }

    public void hold() throws InterruptedException {
        masterThread.join();
    }

    void add(FutureOperate futureOperate){
        queue.add(futureOperate);
    }
}


/*
class WorkerThread extends Thread{
//        final private  FutureOperate[] watchings;

        private final BlockingQueue<FutureOperate> watchings;
        private int busy;
        private int watchingNum;

        public WorkerThread(int watchingNum) {
            this.watchingNum = watchingNum;
//            this.watchings = new FutureOperate[watchingNum];
            this.watchings = new LinkedBlockingQueue<>(watchingNum);
        }

        //不能阻塞
        boolean add(FutureOperate futureOperate){
            return watchings.offer(futureOperate);

        }
//        void add(FutureOperate futureOperate){
//            for (int i = 0; i < watchings.length; i++) {
//                if (watchings[i]==null){
//                    watchings[i]=futureOperate;
//                    busy++;
//                    return;
//                }
//
//            }
//        }

        @Override
        public void run() {
            for (;;){
                for (int i = 0;i<watchings.length;i++){
                    FutureOperate fo = watchings[i];
                    if (fo==null)
                        continue;
                    if (!fo.finished()){
                        fo.work(null);
                    }
                }
            }
        }

        public void free(){
            for (int i = 0;i<watchings.length;i++){
                FutureOperate fo = watchings[i];
                if (fo==null)
                    continue;

                if (fo.finished()){

                    watchings[i]=null;
                    fo.onFinished();
                    busy--;
                }
            }
        }
    }
*/